package com.atguigu.gmall.realtime.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import org.codehaus.plexus.util.StringUtils;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * @author: xu
 * @desc: 用于维度查询的工具类，底层调用的是PhoenixUtil
 * select * from dim_base_trademark where id=10 and name=zs;
 */
public class DimUtil {
    /**
     * 在做维度关联的时候，大部分场景都是通过id进行关联，所以提供一个方法，只需要将id的值作为参数传进来即可
     */
    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName, Tuple2.of("id", id));
    }

    /*
      TODO：优化：查询数据时先从Redis查询，如果Redis没有再到Phoenix查询，并将查询结果放到Redis缓存中
       redis
            类型：    string
            Key：    dim:表名:值   例如：dim:DIM_BASE_TRADEMARK:主键1_主键2_xxx
            value：  通过PhoenixUtil到维度表中查询数据，取出第一条并将其转换为json字符串
            失效时间:  24*3600
        "DIM_BASE_TRADEMARK", Tuple2.of("id", "13"), Tuple2.of("tm_name","zz"))
        redisKey = "dim:dim_base_trademark:13_zz"
    */
    @SafeVarargs
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... cloNameAndValue) {
        // 拼接查询维度SQL
        StringBuilder selectSql = new StringBuilder("select * from " + tableName + " where ");
        // 拼接查询Redis的key
        StringBuilder redisKey = new StringBuilder("dim:" + tableName.toLowerCase() + ":");
        for (int i = 0; i < cloNameAndValue.length; i++) {
            Tuple2<String, String> tuple2 = cloNameAndValue[i];
            String filedName = tuple2.f0;
            String fieldValue = tuple2.f1;
            if (i > 0) {
                selectSql.append(" and ");
                redisKey.append("_");
            }
            selectSql.append(filedName).append("='").append(fieldValue).append("'");
            redisKey.append(fieldValue);
        }
        // 声明操作Redis的客户端
        Jedis jedis;
        // 声明变量，用于接收从Redis中查询出的缓存数据
        String dimJsonStr;
        // 声明变量，用于处理返回的维度对象
        JSONObject dimJsonObj = null;
        try {
            // 获取jedis客户端
            jedis = RedisUtil.getJedis();
            // 根据key到Redis中查询
            dimJsonStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("从redis中查询维度失败");
        }
        // 判断是否从Redis中查询到了数据
        if (StringUtils.isNotBlank(dimJsonStr)) {
            // 从Redis中查到了数据，将缓存的维度字符串转换为json对象
            dimJsonObj = JSON.parseObject(dimJsonStr);
        } else {
            // 如果在Redis中没有查到数据，需要到Phoenix中查询
            System.out.println("查询维度的SQL:" + selectSql);
            List<JSONObject> dimList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            // 对于维度查询来讲，一般都是根据主键进行查询，不可能返回多条记录，只会有一条
            if (dimList.size() > 0) {
                dimJsonObj = dimList.get(0);
                // 将查询出来的数据放到Redis中缓存起来
                jedis.setex(redisKey.toString(), 3600 * 24, dimJsonObj.toJSONString());
            } else {
                System.out.println("维度数据没有找到:" + selectSql);
            }
        }
        // 关闭Jedis
        jedis.close();
        System.out.println("----关闭Redis连接----");
        return dimJsonObj;
    }

    /**
     * 根据key让Redis中的缓存失效
     */
    public static void deleteCached(String tableName, String id) {
        String key = "dim:" + tableName.toLowerCase() + ":" + id;
        try {
            Jedis jedis = RedisUtil.getJedis();
            // 通过key清除缓存
            jedis.del(key);
            jedis.close();
        } catch (Exception e) {
            System.out.println("删除Redis缓存发生了异常！");
            e.printStackTrace();
        }
    }

    /**
     * 从Phoenix中查询数据，没有使用缓存
     */
    @SafeVarargs
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... cloNameAndValue) {
        // 拼接查询条件
        StringBuilder selectSql = new StringBuilder("select * from " + tableName + " where ");
        for (int i = 0; i < cloNameAndValue.length; i++) {
            Tuple2<String, String> tuple2 = cloNameAndValue[i];
            String filedName = tuple2.f0;
            String fieldValue = tuple2.f1;
            if (i > 0) {
                selectSql.append(" and ");
            }
            selectSql.append(filedName).append("='").append(fieldValue).append("'");
        }
        System.out.println("查询维度的SQL:" + selectSql);

        List<JSONObject> dimList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        JSONObject dimJsonObj = null;
        // 对于维度查询来讲，一般都是根据主键进行查询，不可能返回多条记录，只会有一条
        if (dimList.size() > 0) {
            dimJsonObj = dimList.get(0);
        } else {
            System.out.println("维度数据没有找到:" + selectSql);
        }
        return dimJsonObj;
    }

    public static void main(String[] args) {
        // System.out.println(PhoenixUtil.queryList("select * from DIM_BASE_TRADEMARK", JSONObject.class));
        // JSONObject dimInfo = DimUtil.getDimInfoNoCache("DIM_BASE_TRADEMARK", Tuple2.of("id", "14"));

        JSONObject dimInfo = DimUtil.getDimInfo("DIM_BASE_TRADEMARK", "12");
        System.out.println(dimInfo);
    }
}
